package com.zf.flink.simple;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;

import java.util.Objects;

public class MysqlToRedisSync {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(10_000); // 每10秒检查点

        // 构建 MySQL CDC Source
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
            .hostname("localhost")
            .port(3306)
            .databaseList("bobdb")
            .tableList("user") // 监听的表
            .username("root")
            .password("123456")
            .deserializer(new JsonDebeziumDeserializationSchema()) // 输出 JSON 字符串
            .build();

        // 读取数据流
        env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source")
           .map(recordJson -> {
               // 解析 JSON，提取 after 字段（更新后的数据）
               ObjectMapper mapper = new ObjectMapper();
               JsonNode node = mapper.readTree(recordJson);
               JsonNode after = node.get("after");
               return after != null ? after.toString() : null;
           })
           .filter(Objects::nonNull)
           .addSink(new RedisSink<>(new FlinkJedisPoolConfig.Builder().setHost("localhost").setPort(6379).build(),
                   new UserRedisMapper()));

        env.execute("MySQL to Redis Sync");
    }

    // 自定义 Redis 写入逻辑
    public static class UserRedisMapper implements RedisMapper<String> {

        @Override
        public RedisCommandDescription getCommandDescription() {
            return new RedisCommandDescription(RedisCommand.HSET, "user_cache"); // 使用 Hash 存储
        }

        @Override
        public String getKeyFromData(String json) {
            // 从 JSON 提取主键，如 id
            try {
                JsonNode node = new ObjectMapper().readTree(json);
                return node.get("id").asText();
            } catch (Exception e) {
                return "unknown";
            }
        }

        @Override
        public String getValueFromData(String json) {
            return json; // 存储完整对象 JSON
        }
    }
}